#include <errno.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>

#include "compat.h"

#include "psd_logger.h"
#include "psd_measurement.h"
#include "psd_session.h"
#include "psd_xfer.h"
#include "psd_modules.h"
#include "psd_backend_handler.h"


void *psd_xfer_sr_thread(void *arg);
int psd_xfer_sr_init();
static int psd_xfer_sr_main(psdXferArgs *p_args, psdXferArgs *c_args, uint64_t xfer_amt);


psdModule psd_xfer_sr_module = {
        .desc = "Send/Recv Module",
        .dependencies = "",
        .init = psd_xfer_sr_init
};

psdBackendHandler psd_xfer_sr_backend_handler = {
	.name = "sr",
	.run = psd_xfer_sr_main
};

psdModule *module_info() {
        return &psd_xfer_sr_module;
}

int psd_xfer_sr_init() {
	return psd_add_backend_handler(&psd_xfer_sr_backend_handler);
}

static int psd_xfer_sr_main(psdXferArgs *p_args, psdXferArgs *c_args, uint64_t xfer_amt) {
	pthread_t thread1, thread2;

	// first allocate our buffers
	p_args->buf = malloc(sizeof(char) * p_args->bufsize);
	if (!p_args->buf)
		goto error_exit;

	c_args->buf = malloc(sizeof(char) * c_args->bufsize);
	if (!c_args->buf)
		goto error_exit;

        // create an I/O thread
	pthread_mutex_lock(&(p_args->lock));
        if (pthread_create(&thread1, NULL, psd_xfer_sr_thread, (void *) p_args) < 0) {
                pthread_mutex_unlock(&(p_args->lock));
                goto error_exit;
        }

        // wait for the child to initialize
        pthread_cond_wait(&(p_args->cond), &(p_args->lock));
        pthread_mutex_unlock(&(p_args->lock));

        // create an I/O thread
        pthread_mutex_lock(&(c_args->lock));
        if (pthread_create(&thread2, NULL, psd_xfer_sr_thread, (void *) c_args) < 0) {
                pthread_mutex_unlock(&(c_args->lock));
                goto error_exit1;
        }
        // wait for the child to initialize
        pthread_cond_wait(&(c_args->cond), &(c_args->lock));
        pthread_mutex_unlock(&(c_args->lock));

        // setup the threads to transfer the specified amount
        pthread_mutex_lock(&(p_args->lock));
        pthread_mutex_lock(&(c_args->lock));
        {
                p_args->xfer_amt = xfer_amt;
                pthread_cond_signal(&(p_args->cond));

                c_args->xfer_amt = xfer_amt;
                pthread_cond_signal(&(c_args->cond));
        }
        pthread_mutex_unlock(&(p_args->lock));
        pthread_mutex_unlock(&(c_args->lock));
	
	// wait for the I/O threads to finish
	pthread_join(thread1, NULL);
        pthread_join(thread2, NULL);

	return 0;

 error_exit1:
	pthread_cancel(thread1);
 error_exit:
	return -1;
}


void *psd_xfer_sr_thread(void *arg) {
        psdXferArgs *xfer_args = arg;
	psdConn *curr_src = xfer_args->src;
	psdConn *curr_sink = xfer_args->sink;
	uint64_t sent;
	uint64_t bytes_sent;
	size_t bytes_sent_size;
	struct timeval start_time, end_time;
	char filename[255];
//	FILE *output;

#ifdef NETLOGGER
        unsigned long long block_id = 0;
	char *guid;
	int s_id;

	NL_get_guid(&guid);
#endif

	pthread_mutex_lock(&(xfer_args->lock));

	psd_info(2, "new transfer thread: %s->%s", curr_src->description, curr_sink->description);

/*
	sprintf(filename, "output.%s-%s", curr_src->description, curr_sink->description);
	{
		int i;
		for(i = 0; i < strlen(filename); i++) {
			if (filename[i] == '/') {
				filename[i] = '_';
			}
		}
	}

	output = fopen(filename, "w");
	*/

	gettimeofday(&start_time, NULL);

	pthread_cond_signal(&(xfer_args->cond)); // alert the parent that we're running and about to wait for them to tell us how much data to transfer
	
	while (1) {
	        pthread_cond_wait(&(xfer_args->cond), &(xfer_args->lock));

		if (xfer_args->xfer_amt == (uint64_t) -1) {
			psd_info(2, "%s->%s: going into perpetual receive mode", curr_src->description, curr_sink->description);
		} else {
			psd_info(2, "%s->%s: reciving %llu bytes", curr_src->description, curr_sink->description, xfer_args->xfer_amt);
		}

		sent = 0;

		while(sent < xfer_args->xfer_amt) {
			int bytes_left;
			int bytes_sent;
			int recv_amt;
			int n;

			if (xfer_args->bufsize < (xfer_args->xfer_amt - sent)) {
				recv_amt = xfer_args->bufsize;
			} else {
				recv_amt = xfer_args->xfer_amt - sent; 
			}

			pthread_mutex_lock(&(curr_src->lock));
			{
				while (curr_src->status == STATUS_FROZEN) {
					psd_info(0, "%s->%s: connection %s frozen", curr_src->description, curr_sink->description, curr_src->description);
					pthread_cond_wait(&(curr_src->cond), &(curr_src->lock));
				}
			}
			pthread_mutex_unlock(&(curr_src->lock));

#ifdef NETLOGGER
			guid = curr_src->session->sess_id;
			s_id = curr_src->session->nl_id;
			NL_transfer_start(nllog, NL_LVL_UDEF, NL_TRANSFER_NET_READ, guid, s_id, block_id);
#endif

			psd_info(11, "%s->%s: going into recv");
			n = psd_conn_read(curr_src, xfer_args->buf, recv_amt, 0);
#ifdef NETLOGGER
			NL_transfer_end(nllog, NL_LVL_UDEF, NL_TRANSFER_NET_READ, guid, s_id, block_id, (double)n);
			block_id++;
#endif
			psd_info(11, "%s->%s: recv returned %d", curr_src->description, curr_sink->description, n);

			if (n <= 0) {
				if(n < 0) {
					psd_err(0, "%s->%s: recv failed: %s", curr_src->description, curr_sink->description, strerror(errno));
				}

				goto out;
			}

			//fwrite(xfer_args->buf, sizeof(char), n, output);

			bytes_left = n;
			bytes_sent = 0;

			pthread_mutex_lock(&(curr_sink->lock));
			{
				while (curr_sink->status == STATUS_FROZEN) {
					psd_info(0, "%s->%s: connection %s frozen", curr_src->description, curr_sink->description, curr_sink->description);
					pthread_cond_wait(&(curr_sink->cond), &(curr_sink->lock));
				}
			}
			pthread_mutex_unlock(&(curr_sink->lock));

			while(bytes_left > 0) {
#ifdef NETLOGGER			        
			        guid = curr_sink->session->sess_id;
				s_id = curr_src->session->nl_id;
			        NL_transfer_start(nllog, NL_LVL_UDEF, NL_TRANSFER_NET_WRITE, guid, s_id, block_id);
#endif
				psd_info(11, "%s->%s: going into send");
			        n = psd_conn_write(curr_sink, xfer_args->buf + bytes_sent, bytes_left, 0);
#ifdef NETLOGGER
				NL_transfer_end(nllog, NL_LVL_UDEF, NL_TRANSFER_NET_WRITE, guid, s_id, block_id, (double)n);
			        block_id++;
#endif
				if (n < 0) {
					psd_err(0, "%s->%s: send failed: %s", curr_src->description, curr_sink->description, strerror(errno));
					goto out;
				}

				psd_info(11, "%s->%s: send returned %d", curr_src->description, curr_sink->description, n);

				bytes_left -= n;
				sent += n;
			}
		}

		if (write(xfer_args->parent_fd, "1", 1) < 0) {
			psd_err(0, "%s->%s: couldn't communicate with parent thread: %s", curr_src->description, curr_sink->description, strerror(errno));
			goto out;
		}
	}

out:
	psd_info(1, "%s->%s: closing sockets and returning.", curr_src->description, curr_sink->description);

	gettimeofday(&end_time, NULL);

	bytes_sent_size = sizeof(bytes_sent);
	if (psd_conn_get_stat(curr_sink, PSD_STATS_BYTES_WRITTEN, &bytes_sent, &bytes_sent_size) == 0) {
		char tmp[255];
		snprintf(tmp, sizeof(tmp), "%s/%s->%s", psd_session_get_id(xfer_args->sess), curr_src->description, curr_sink->description);
		psd_log_measurement_uint(end_time, tmp, "TransferBytes", bytes_sent);
		psd_log_measurement_dbl(end_time, tmp, "TransferRate", ((bytes_sent/difftv(&start_time, &end_time))*8)/1000/1000);
		psd_log_measurement_dbl(end_time, tmp, "TransferTime", difftv(&start_time, &end_time));
	}

/*
	fclose(output);
	*/

	psd_conn_shutdown(curr_sink, PSD_SEND_SIDE);
	psd_conn_shutdown(curr_src, PSD_RECV_SIDE);
	
	pthread_exit(NULL);
}
